package chapter06

import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable
/**
 * author: 余辉
 * blog: https://blog.csdn.net/silentwolfyh
 * date: 2024 - 09 - 02 11:15 上午
 *
 * 需求：
 * 自定义累加器，用于统计单词出现的次数
 */

object MyAccDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("wc")
    val sc = new SparkContext(conf)

    val rdd = sc.makeRDD(List("hello spark", "hello scala", "hello spark", "spark scala"))
    // 声明自定义累加器
    val sum = new MyWordCountAcc
    // 注册累加器到 Spark
    sc.register(sum, "WordCountAcc")

    // 处理 RDD，统计单词
    rdd.flatMap(_.split(" ")).foreach(word => sum.add(word))

    // 在任务完成后打印累加器的值
    println(sum.value)

    sc.stop()
  }
}

// 自定义累加器，用于统计单词出现的次数
class MyWordCountAcc extends AccumulatorV2[String, mutable.Map[String, Int]] {
  // 使用可变映射来存储单词及其计数，初始值为0
  protected val wordMap = mutable.Map[String, Int]().withDefaultValue(0)

  // 检查累加器是否处于“零”状态，即没有单词被添加
  override def isZero: Boolean = wordMap.isEmpty

  // copy 方法应该返回一个新的累加器实例，而不是复制当前状态
  // 这里的实现是错误的，因为它试图在匿名子类中访问受保护的 wordMap
  // 正确的做法应该是返回一个新的 MyWordCountAcc 实例，可能是一个空实例
  // 但由于 AccumulatorV2 的设计，我们通常不需要在 copy 方法中复制状态
  // 因此，下面是一个修正后的实现，它只返回一个新的空累加器实例
  override def copy(): AccumulatorV2[String, mutable.Map[String, Int]] = new MyWordCountAcc

  // 重置累加器的状态，清空单词映射
  override def reset(): Unit = wordMap.clear()

  // 向累加器中添加一个单词，并增加其计数
  override def add(word: String): Unit = {
    wordMap(word) += 1
  }

  // 合并另一个累加器的状态到当前累加器中
  // 注意：这里的实现应该只读取 other.value 并更新 wordMap，而不是修改 other
  override def merge(other: AccumulatorV2[String, mutable.Map[String, Int]]): Unit = {
    other.value.foreach { case (word, count) => wordMap(word) += count }
  }

  // 返回累加器的当前状态，即单词及其计数的映射
  override def value: mutable.Map[String, Int] = wordMap
}